// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: change E2E packet format
// feature: Support Cache fast application and slow release mechanism
// feature: develop dispatcher
// feature: content filter topic
// feature: Split history as an independent common module
// feature: discovery support client and server
// feature: log event record topic name
// feature: add statictic lookup
// feature: E2E Profile04 and Configuration Function Development
// ------------------------------------------------------------------

#ifndef INCLUDE_EDDS_RTPS_READER_RTPSREADER_H_
#define INCLUDE_EDDS_RTPS_READER_RTPSREADER_H_

#include <functional>

#include <history/SubstrateReader.h>
#include <edds/rtps/Endpoint.h>
#include <edds/rtps/attributes/ReaderAttributes.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <edds/rtps/common/SequenceNumber.h>
#include <edds/rtps/common/Time_t.h>
#include <history/ReaderHistory.h>
#include <ertps/qos/LivelinessChangedStatus.h>
#include <ertps/qos/SampleLostStatus.hpp>
#include <ertps/utils/TimedConditionVariable.hpp>
#include <edds/dds/builtin/lookup/common/LookupTypes.hpp>
#include <elog/statistics/Statistics.hpp>
#include <history/ITopicPayloadPool.h>
#include <edds/rtps/interfaces/IReaderDataFilter.hpp>

namespace vbs {
namespace common {
class TimedEvent;
class ReaderListener;
class History;
struct CacheChange_t;
}  // namespace common
}  // namespace vbs
namespace evbs {
namespace ertps {
namespace rtps {

// Forward declarations

struct ReaderHistoryState;
class WriterProxy;
using vbs::common::ReaderHistory;

/**
 * Class RTPSReader, manages the reception of data from its matched writers.
 * @ingroup READER_MODULE
 */
class RTPSReader : public Endpoint,
                   public vbsutil::statistics::StatisticsReader,
                   public vbs::dispatcher::SubstrateReader {
    friend class ReaderHistory;
    friend class RTPSParticipantImpl;
    friend class MessageReceiver;
    friend class EDP;
    friend class WLP;
    using ITopicPayloadPool = vbs::common::ITopicPayloadPool;

 protected:
    RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att, ReaderHistory* hist,
               vbs::common::ReaderListener* listen = nullptr);

    RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
               const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool, ReaderHistory* hist,
               vbs::common::ReaderListener* listen = nullptr);

    RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
               const std::shared_ptr<vbs::common::IChangePool>& change_pool, ReaderHistory* hist,
               vbs::common::ReaderListener* listen = nullptr);

    RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
               const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
               const std::shared_ptr<vbs::common::IChangePool>& change_pool, ReaderHistory* hist,
               vbs::common::ReaderListener* listen = nullptr);

    virtual ~RTPSReader();

 public:
    /**
     * Add a matched writer represented by its attributes.
     * @param wdata Attributes of the writer to add.
     * @return True if correctly added.
     */
    RTPS_DllAPI virtual bool matched_writer_add(WriterProxyData& wdata) = 0;

    /**
     * Remove a writer represented by its attributes from the matched writers.
     * @param writer_guid GUID of the writer to remove.
     * @param removed_by_lease Whether the writer is being unmatched due to a participant drop.
     * @return True if correctly removed.
     */
    RTPS_DllAPI virtual bool matched_writer_remove(const GUID_t& writer_guid, bool removed_by_lease = false) = 0;

    /**
     * Tells us if a specific Writer is matched against this reader.
     * @param writer_guid GUID of the writer to check.
     * @return True if it is matched.
     */
    RTPS_DllAPI virtual bool matched_writer_is_matched(const GUID_t& writer_guid) = 0;

    /**
     * Processes a new DATA message. Previously the message must have been accepted by function
     * acceptMsgDirectedTo.
     *
     * @param change Pointer to the CacheChange_t.
     * @return true if the reader accepts messages from the.
     */
    RTPS_DllAPI virtual bool processDataBatchMsg(CacheChange_t* change) = 0;

    /**
     * Processes a new DATA message. Previously the message must have been accepted by function
     * acceptMsgDirectedTo.
     *
     * @param change Pointer to the CacheChange_t.
     * @return true if the reader accepts messages from the.
     */
    RTPS_DllAPI virtual bool processDataMsg(CacheChange_t* change) = 0;

    /**
     * Processes a new DATA FRAG message.
     *
     * @param change Pointer to the CacheChange_t.
     * @param sampleSize Size of the complete, assembled message.
     * @param fragmentStartingNum Starting number of this particular message.
     * @param fragmentsInSubmessage Number of fragments on this particular message.
     * @return true if the reader accepts message.
     */
    RTPS_DllAPI virtual bool processDataFragMsg(CacheChange_t* change, uint32_t sampleSize,
                                                uint32_t fragmentStartingNum, uint16_t fragmentsInSubmessage) = 0;

    /**
     * Processes a new HEARTBEAT message.
     * @param writerGUID
     * @param hbCount
     * @param firstSN
     * @param lastSN
     * @param finalFlag
     * @param livelinessFlag
     * @return true if the reader accepts messages from the.
     */
    RTPS_DllAPI virtual bool processHeartbeatMsg(const GUID_t& writerGUID, uint32_t hbCount,
                                                 const SequenceNumber_t& firstSN, const SequenceNumber_t& lastSN,
                                                 bool finalFlag, bool livelinessFlag) = 0;

    /**
     * Processes a new GAP message.
     * @param writerGUID
     * @param gapStart
     * @param gapList
     * @return true if the reader accepts messages from the.
     */
    RTPS_DllAPI virtual bool processGapMsg(const GUID_t& writerGUID, const SequenceNumber_t& gapStart,
                                           const SequenceNumberSet_t& gapList) = 0;

    /**
     * Processes a new GAP message.
     * @param writerGUID
     * @param gapStart
     * @param gapList
     * @return true if the reader accepts messages from the.
     */
    RTPS_DllAPI virtual bool processE2EMsg(const E2EProfile04_t& e2eHeader) = 0;

    RTPS_DllAPI virtual edds::dds::builtin::StatisticMatchGuids get_remote_guids() {
        return edds::dds::builtin::StatisticMatchGuids();
    };

    RTPS_DllAPI virtual edds::dds::builtin::StatisticProxyInfos get_proxy_infos() {
        return edds::dds::builtin::StatisticProxyInfos();
    };  // LCOV_EXCL_STOP

    /**
     * Get the associated listener, secondary attached Listener in case it is of compound type
     * @return Pointer to the associated reader listener.
     */
    RTPS_DllAPI vbs::common::ReaderListener* getListener() const;

    /**
     * Switch the ReaderListener kind for the Reader.
     * If the RTPSReader does not belong to the built-in protocols it switches out the old one.
     * If it belongs to the built-in protocols, it sets the new ReaderListener callbacks to be
     * called after the built-in ReaderListener ones.
     * @param target Pointed to ReaderLister to attach
     * @return True is correctly set.
     */
    //user writer is't able to call this func
    RTPS_DllAPI bool setListener(vbs::common::ReaderListener* target);

    /**
     * Reserve a CacheChange_t.
     * @param change Pointer to pointer to the Cache.
     * @param dataCdrSerializedSize Size of the Cache.
     * @return True if correctly reserved.
     */
    RTPS_DllAPI bool reserveCache(CacheChange_t** change, uint32_t dataCdrSerializedSize);

    /**
     * Release a cacheChange.
     */
    RTPS_DllAPI void releaseCache(CacheChange_t* change);

    /**
     * Get the next CacheChange_t from the history to take.
     * @param change Pointer to pointer of CacheChange_t.
     * @param wp Pointer to pointer to the WriterProxy.
     * @return True if read.
     */
    RTPS_DllAPI virtual bool nextUntakenCache(CacheChange_t** change, WriterProxy** wpout) = 0;

    RTPS_DllAPI bool wait_for_unread_cache(const evbs::ertps::Duration_t& timeout);

    RTPS_DllAPI uint64_t get_unread_count() const;

    RTPS_DllAPI uint64_t get_unread_count(bool mark_as_read);

    //! Returns a pointer to the associated History.
    RTPS_DllAPI inline ReaderHistory* getHistory() const { return mp_history; }

    /**
     * @return The content filter associated to this reader.
     */
    RTPS_DllAPI inline IReaderDataFilter* get_content_filter() {
        std::lock_guard<vbs::common::RecursiveTimedMutex> lock(mp_mutex);
        return data_filter_;
    }

    /**
     * Set the content filter associated to this reader.
     *
     * @param filter  Pointer to the content filter to associate to this reader.
     */
    RTPS_DllAPI inline void reader_data_filter(IReaderDataFilter* filter) {
        std::lock_guard<vbs::common::RecursiveTimedMutex> lock(mp_mutex);
        data_filter_ = filter;
    }

    RTPS_DllAPI bool parseChangeFromBatch(CacheChange_t* change_to_add, CacheChange_t* change_batch,
                                          uint32_t& sample_pos, uint32_t& data_pos);

    //! The liveliness changed status struct as defined in the DDS
    LivelinessChangedStatus liveliness_changed_status_;

    inline void enableMessagesFromUnkownWriters(const bool enable) { m_acceptMessagesFromUnkownWriters = enable; }

    void setTrustedWriter(const EntityId_t& writer) {
        m_acceptMessagesFromUnkownWriters = false;
        m_trustedWriterEntityId = writer;
    }

    virtual size_t getMatchedWritersSize() const = 0;

    /**
     * Assert the liveliness of a matched writer.
     * @param writer GUID of the writer to assert.
     */
    virtual void assert_writer_liveliness(const GUID_t& writer) const = 0;

    /**
     * @brief Check if this reader is in a clean state with all its matched writers.
     * This will happen when the reader has received all samples announced by all its matched writers.
     *
     * @return Whether the reader is in a clean state with all its matched writers.
     */
    virtual bool is_in_clean_state() = 0;

    /**
     * Called just before a change is going to be deserialized.
     * @param [in]  change            Pointer to the change being accessed.
     * @param [out] wp                Writer proxy the @c change belongs to.
     * @param [out] is_future_change  Whether the change is in the future (i.e. there are
     *                                earlier unreceived changes from the same writer).
     *
     * @return Whether the change is still valid or not.
     */
    virtual bool begin_sample_access_nts(CacheChange_t* change, bool& is_future_change) override {  // LCOV_EXCL_START
        UNUSED_PARAMETER(change);
        UNUSED_PARAMETER(is_future_change);
        return true;
    }  // LCOV_EXCL_STOP

    /**
     * Called after the change has been deserialized.
     * @param [in] change        Pointer to the change being accessed.
     * @param [in] wp            Writer proxy the @c change belongs to.
     * @param [in] mark_as_read  Whether the @c change should be marked as read or not.
     */
    virtual void end_sample_access_nts(CacheChange_t* change, /*WriterProxy*& wp,*/ bool mark_as_read) = 0;

    /**
     * Called when the user has retrieved a change from the history.
     * @param change Pointer to the change to ACK
     * @param writer Writer proxy of the \c change.
     * @param mark_as_read Whether the \c change should be marked as read or not
     */
    virtual void change_read_by_user(CacheChange_t* const change, const WriterProxy* const writer,
                                     const bool mark_as_read = true) = 0;

    /**
     * Checks whether the sample is still valid or is corrupted.
     *
     * @param data    Pointer to the sample data to check.
     *                If it does not belong to the payload pool passed to the
     *                reader on construction, it yields undefined behavior.
     * @param writer  GUID of the writer that sent \c data.
     * @param sn      Sequence number related to \c data.
     *
     * @return true if the sample is valid
     */
    RTPS_DllAPI bool is_sample_valid(const void* data, const GUID_t& writer, const SequenceNumber_t& sn) const;

    const std::string& getTopicName() const override { /* override */
        return m_topicName;
    }

    void setTopicName(const std::string topicName) { m_topicName = topicName; }

    std::string getIdlCrc16() const { return m_crc16Idl; }

    void setIdlCrc16(const std::string crc16_input) { m_crc16Idl = crc16_input; }

    //! get is_send_multi
    bool getSendMulti() const { return is_send_multi; }

    //! set is_send_multi
    void setSendMulti(bool send_multi) { is_send_multi = send_multi; }

    void add_announce_locator(Locator_t loc) override;
    inline RTPSParticipantImpl* getRTPSParticipant() const { return mp_RTPSParticipant; }

    uint8_t getTracePointStatus() const { return trace_point_status_; }

    ReaderHistoryState* getHistoryState() { return history_state_; }

 protected:
    virtual bool may_remove_history_record(bool removed_by_lease);

    /*!
     * @brief Add a remote writer to the persistence_guid map
     * @param guid GUID of the remote writer
     * @param persistence_guid Persistence GUID of the remote writer
     */
    void add_persistence_guid(const GUID_t& guid, const GUID_t& persistence_guid);

    /*!
     * @brief Remove a remote writer from the persistence_guid map
     * @param guid GUID of the remote writer
     * @param persistence_guid Persistence GUID of the remote writer
     * @param removed_by_lease Whether the GUIDs are being removed due to a participant drop.
     */
    void remove_persistence_guid(const GUID_t& guid, const GUID_t& persistence_guid, bool removed_by_lease);

    /*!
     * @brief Get the last notified sequence for a RTPS guid
     * @param guid The RTPS guid to query
     * @return Last notified sequence number for input guid
     * @remarks Takes persistence_guid into consideration
     */
    SequenceNumber_t get_last_notified(const GUID_t& guid);

    /*!
     * @brief Update the last notified sequence for a RTPS guid
     * @param guid The RTPS guid of the writer
     * @param seq Max sequence number available on writer
     * @return Previous value of last notified sequence number for input guid
     * @remarks Takes persistence_guid into consideration
     */
    SequenceNumber_t update_last_notified(const GUID_t& guid, const SequenceNumber_t& seq);

    /*!
     * @brief Set the last notified sequence for a persistence guid
     * @param persistence_guid The persistence guid to update
     * @param seq Sequence number to set for input guid
     * @remarks Persistent readers will write to DB
     */
    virtual void set_last_notified(const GUID_t& persistence_guid, const SequenceNumber_t& seq);

    /*!
     * @brief Search if there is a CacheChange_t, giving SequenceNumber_t and writer GUID_t,
     * waiting to be completed because it is fragmented.
     * @param sequence_number SequenceNumber_t of the searched CacheChange_t.
     * @param writer_guid writer GUID_t of the searched CacheChange_t.
     * @param change If a CacheChange_t was found, this argument will fill with its pointer.
     * In other case nullptr is returned.
     * @param hint Iterator since the search will start.
     * Used to improve the search.
     * @return Iterator pointing to the position were CacheChange_t was found.
     * It can be used to improve next search.
     */
    vbs::common::History::const_iterator findCacheInFragmentedProcess(const SequenceNumber_t& sequence_number,
                                                                      const GUID_t& writer_guid, CacheChange_t** change,
                                                                      vbs::common::History::const_iterator hint) const;

    //! ReaderHistory
    ReaderHistory* mp_history;
    //! Listener
    vbs::common::ReaderListener* mp_listener;
    //! Accept msg to unknwon readers (default=true)
    bool m_acceptMessagesToUnknownReaders;
    //! Accept msg from unknwon writers (BE-true,RE-false)
    bool m_acceptMessagesFromUnkownWriters;
    //! Trusted writer (for Builtin)
    EntityId_t m_trustedWriterEntityId;

    //! ReaderHistoryState
    ReaderHistoryState* history_state_;

    //uint64_t total_unread_ = 0U;

    TimedConditionVariable new_notification_cv_;

    //! The liveliness kind of this reader
    LivelinessQosPolicyKind liveliness_kind_;
    //! The liveliness lease duration of this reader
    Duration_t liveliness_lease_duration_;

    //! Whether sending multiple copies is supported
    bool is_send_multi = false;
    //! Whether builtin writer.
    bool builtin = false;
    //! Enable E2E Protection
    bool enable_e2e_protection_;

    //! E2E Profile04
    uint16_t e2e_p04_min_data_length_;
    uint16_t e2e_p04_max_data_length_;
    uint16_t e2e_p04_max_delta_counter_;

    //! E2E Profile 04 information
    E2EProfile04_t e2e_profile04_;

    uint32_t mem_watermark = 0;

    uint8_t trace_point_status_;

 private:
    RTPSReader& operator=(const RTPSReader&) = delete;

    void init(const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
              const std::shared_ptr<vbs::common::IChangePool>& change_pool, const ReaderAttributes& att);

    std::shared_ptr<vbs::common::IPayloadPool> get_payload_pool(ReaderHistory* hist);
    void release_payload_pool();
    bool history_adjustment();

    std::string m_crc16Idl;

    std::string m_topicName;

    IReaderDataFilter* data_filter_ {};
    vbs::common::TimedEvent* free_event_ = nullptr;
};

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */

#endif  // INCLUDE_EDDS_RTPS_READER_RTPSREADER_H_
